package com.zf.flink;// com/example/sink/RedisSink.java
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

public class RedisSink extends RichSinkFunction<User> {

    private transient Jedis jedis;

    @Override
    public void open(Configuration parameters) {
        jedis = new Jedis("localhost", 6379);
        jedis.connect();
    }

    @Override
    public void invoke(User user, Context context) {
        System.out.printf("User: %s\n", user);
        String key = "user:" + user.getId();
        jedis.hset(key, "name", user.getName());
        jedis.hset(key, "age", user.getAge().toString());
        jedis.expire(key, 3600); // 1小时过期
    }

    @Override
    public void close() {
        if (jedis != null && jedis.isConnected()) {
            jedis.close();
        }
    }
}